package com.crpc.core.detector;

import com.crpc.common.crpcEnum.CrpcExceptionEnums;
import com.crpc.common.crpcEnum.RequestType;
import com.crpc.common.exception.CrpcException;
import com.crpc.core.CrpcBootStrap;
import com.crpc.core.NettyBootstrapInitializer;
import com.crpc.core.compress.CompressorFactory;
import com.crpc.core.discovery.Registry;
import com.crpc.core.serialize.SerializerFactory;
import com.crpc.core.transport.message.CrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static com.crpc.core.CrpcBootStrap.ANSWER_TIME;
import static com.crpc.core.CrpcBootStrap.CHANNEL_MAP;

/**
 * @author wzy
 * @version 1.0
 * @description: 心跳检测
 * @date 2023/10/1 14:31
 */
@Slf4j
public class HeartbeatDetector {
    public static void detectorHeartbeat(String serviceName) {
        //从注册中心拉取服务列表并建立连接
        //将连接进行缓存
        Registry registry = CrpcBootStrap.getInstance().getConfiguration().getRegistryConfig().getRegistry();
        List<InetSocketAddress> inetSocketAddressList = registry.lookup(serviceName);
        for (InetSocketAddress inetSocketAddress : inetSocketAddressList) {
            try {
                if (!CHANNEL_MAP.containsKey(inetSocketAddress)) {
                    Channel sync = NettyBootstrapInitializer.getBootstrap().connect(inetSocketAddress).sync().channel();
                    CHANNEL_MAP.put(inetSocketAddress, sync);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
            }
        }
        //循环定时任务发送消息
        Thread thread = new Thread(() -> {
            new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 15000);
        }, "心跳检测线程");
        thread.setDaemon(true);
        thread.start();
    }

    private static class MyTimerTask extends TimerTask {
        @Override
        public void run() {
            ANSWER_TIME.clear();
            //遍历所以的channel
            for (Map.Entry<InetSocketAddress, Channel> entrySet : CHANNEL_MAP.entrySet()) {
                int tryNum = 3;
                while (tryNum > 0) {
                    Channel channel = entrySet.getValue();
                    long startTime = System.currentTimeMillis();
                    //心跳请求
                    CrpcRequest heartBeatRequest = CrpcRequest.builder()
                            .requestId(CrpcBootStrap.getInstance().getConfiguration().getIdGenerator().getId())
                            .compressType(CompressorFactory.getCompressor(CrpcBootStrap.getInstance().getConfiguration().getCompressorType()).getCode())
                            .requestType(RequestType.HEART_BEAT.getId())
                            .serializeType(SerializerFactory.getSerializer(CrpcBootStrap.getInstance().getConfiguration().getSerializerType()).getCode())
                            .timeStamp(startTime)
                            .build();
                    CompletableFuture<Object> future = new CompletableFuture<>();
                    CrpcBootStrap.PENDING_REQUEST.put(heartBeatRequest.getRequestId(), future);
                    channel.writeAndFlush(heartBeatRequest).addListener((ChannelFutureListener) promise -> {
                        if (!promise.isSuccess()) {
                            future.completeExceptionally(promise.cause());
                        }
                    });
                    long endTime = 0L;
                    try {
                        future.get(1, TimeUnit.SECONDS);
                        endTime = System.currentTimeMillis();
                    } catch (Exception e) {
                        tryNum--;
                        log.error("和{}连接的主机发生异常,剩余重试次数{}", channel.remoteAddress(), tryNum);
                        if (tryNum == 0) {
                            //将失效的地址移除服务列表
                            CHANNEL_MAP.remove(entrySet.getKey());
                        }
                        try {
                            Thread.sleep(10L * (3 - tryNum));
                        } catch (InterruptedException ex) {
                            e.printStackTrace();
                            throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
                        }
                        continue;
                    }
                    //响应时间
                    Long time = endTime - startTime;
                    ANSWER_TIME.put(time, channel);
                    break;
                }
            }
            log.info("==========================响应时间treeMap========心跳检测=================================");
            for (Map.Entry<Long, Channel> entry : ANSWER_TIME.entrySet()) {
                System.out.println("时间:" + entry.getKey() + "----->" + "地址" + entry.getValue().remoteAddress());
            }
        }
    }
}
